package edu.hit.crawler;

import java.io.IOException;
import java.net.URL;
import java.text.SimpleDateFormat;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.apache.log4j.Logger;

import edu.hit.crawler.io.CrawItem;
import edu.hit.crawler.util.URLCanonicalizer;
import edu.hit.crawler.util.URLFilter;

@SuppressWarnings("deprecation")
public class UploadForbiddenHost extends Configured implements Tool {

	public static Logger logger = Logger.getLogger(UploadForbiddenHost.class);

	public static class UploadMapper extends
			Mapper<LongWritable, Text, Text, CrawItem> {
		public void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {

			if (null == value || 0 == value.getLength())
				return;
			String url =  URLCanonicalizer.getCanonicalURL("http://" + value.toString());
			if (null == url || url.length() == 0)
				return;
			context.write(value, new CrawItem(url, CrawItem.FORBIDDEN));
		}
	}

	public static class UploadReducer extends
			Reducer<Text, CrawItem, Text, CrawItem> {

		@Override
		public void reduce(Text key, Iterable<CrawItem> values, Context context)
				throws IOException, InterruptedException {
			for (CrawItem value : values) {
				context.write(key, value);
				break;
			}

		}
	}

	/**
	 * @param args
	 * @throws Exception
	 */
	public static void main(String[] args) throws Exception {
		// TODO Auto-generated method stub
		int res = ToolRunner.run(new Configuration(),
				new UploadForbiddenHost(), args);
		System.exit(res);
	}

	@Override
	public int run(String[] args) throws Exception {
		Configuration conf = getConf();
		Job job = new Job(conf, "Upload");
		job.setJarByClass(Crawler.class);
		job.setMapperClass(UploadMapper.class);
		job.setReducerClass(UploadReducer.class);

		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(CrawItem.class);
		job.setOutputFormatClass(SequenceFileOutputFormat.class);

		job.setNumReduceTasks(2);

		String workdir = conf.get("org.work.crawler.dir", "crawler/");
		FileSystem fs = FileSystem.get(getConf());
		fs.copyFromLocalFile(new Path("forbidden_host_out"), new Path(workdir
				+ "upload/forbidden"));
		FileInputFormat.addInputPath(job, new Path(workdir + "upload"));
		Path out = new Path(workdir + "forbidden/");

		SequenceFileOutputFormat.setOutputPath(job, out);

		int res = job.waitForCompletion(true) ? 0 : 1;

		if (res == 0) {
			System.out.println("Upload done!");
		} else {
			System.out.println("Upload failed.");
		}

		return res;
	}

}
